package com.atguigu.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @author gmd
 * @desc 从kafka读取数据
 * @since 2024-11-24 10:14:18
 */
public class KafkaReadData {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从Kafka中读取数据
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("172.16.0.89:9092") // 指定kafka节点的地址和端口
                .setGroupId("atguigu")  // 指定消费者组的id
                .setTopics("testTopic01")   // 指定消费的 Topic
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定 反序列化器，这个是反序列化value
                .setStartingOffsets(OffsetsInitializer.latest())  // flink消费kafka的策略
                .build();

        env
                // 不生成水印，用于测试
//                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource")
                // 生成水印，最大乱序时间窗口为3秒
                .fromSource(kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkaSource")
                .print();

        env.execute();
    }

}

/*
 * Kafka消费者参数说明：
 *   - auto.reset.offsets：消费者 Offset 重置策略
 *       - earliest：如果有 Offset，从 Offset 继续消费；如果没有 Offset，从最早的消息开始消费
 *       - latest：如果有 Offset，从 Offset 继续消费；如果没有 Offset，从最新的消息开始消费
 *
 * Flink KafkaSource 的 Offset 消费策略：
 *   - OffsetsInitializer：默认为 earliest
 *       - earliest：一定从最早的消息开始消费
 *       - latest：一定从最新的消息开始消费
 *
 * 说明：Flink KafkaSource 的 Offset 消费策略与 Kafka 消费者参数的 auto.reset.offsets 有所不同。
 *       Flink 的 earliest 策略总是从最早的消息开始消费，而 Kafka 的 earliest 策略则是如果有 Offset，从 Offset 继续消费。
 */